package com.tm.netty;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.tm.menum.TransactionType;
import com.tm.transactional.GlobalTransactionManager;
import com.tm.transactional.MyTransaction;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @author laowu
 * @date 2020/8/22 13:51
 * @desc
 */
public class NettyClientHandler extends ChannelInboundHandlerAdapter {

    private ChannelHandlerContext context;
    private Logger log = LoggerFactory.getLogger(NettyClientHandler.class);
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        context = ctx;
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        log.info("接收数据 = {}", msg.toString());
        JSONObject obj = JSON.parseObject((String)msg);
        String xid = obj.getString("xid");
        String command = obj.getString("command"); //commit, rollback
        log.info("接收command = {}", command);
        // 对事务进行操作
        MyTransaction myTransaction = GlobalTransactionManager.getMyTransaction(xid);
        if(myTransaction == null) {
            log.info("未启动分布式事务");
            return;
        }
        if("commit".equals(command)) {
            myTransaction.setTransactionType(TransactionType.commit);
        } else {
            myTransaction.setTransactionType(TransactionType.rollback);
        }
        try {
            myTransaction.getTask().getLock().lock();
            myTransaction.getTask().getCondition().signal();
        }finally {
            myTransaction.getTask().getLock().unlock();
        }
    }

    public synchronized Object call(JSONObject data) {
        context.writeAndFlush(data.toJSONString()).channel().newPromise();
        return null;
    }
}
